package com.bff.gaia.mix.api.xjoin2.datastream;

import com.bff.gaia.api.common.functions.InvalidTypesException;
import com.bff.gaia.api.common.typeinfo.TypeInformation;
import com.bff.gaia.api.java.typeutils.GenericTypeInfo;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.mix.api.datastream.DataMixStream;
import com.bff.gaia.mix.api.datastream.SingleOutputMixStreamOperator;
import com.bff.gaia.mix.api.environment.MixStreamExecutionEnvironment;
import com.bff.gaia.mix.api.xjoin2.core.enums.AsyncXjoinCacheType;
import com.bff.gaia.mix.api.xjoin2.core.enums.XjoinMethod;
import com.bff.gaia.mix.api.xjoin2.core.enums.XjoinType;
import com.bff.gaia.mix.api.xjoin2.core.side.SideTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.FieldInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.StreamTableInfo;
import com.bff.gaia.mix.api.xjoin2.core.table.JoinInfo;
import com.bff.gaia.mix.api.xjoin2.operators.AsyncXjoinOperatorGateWay;
import com.bff.gaia.mix.api.xjoin2.operators.SyncXjoinOperatorGateWay;
import com.bff.gaia.types.Row;

import java.util.ArrayList;
import java.util.List;

import static java.util.Objects.requireNonNull;

public class XjoinMixStreamV4 {
	protected final MixStreamExecutionEnvironment environment;
	private final DataMixStream<Row> streamTable;
	private StreamTableInfo streamTableInfo;
	private SideTableInfo sideTableInfo;
	private List<FieldInfo> outFieldInfoList; //输出列

	public XjoinMixStreamV4(MixStreamExecutionEnvironment environment, DataMixStream streamTable, SideTableInfo sideTableInfo) {
		this.environment = environment;
		this.streamTable = streamTable;
		this.sideTableInfo = sideTableInfo;
		this.streamTableInfo = null;

		TypeInformation typeInformation = streamTable.getType();
		if (typeInformation instanceof GenericTypeInfo) {
			throw new InvalidTypesException("need row type");
		} else {
			RowTypeInfo streamTableRowInfo = (RowTypeInfo) typeInformation;
			this.streamTableInfo = new StreamTableInfo();
			for (int i = 0; i < streamTableRowInfo.getArity(); i++) {
				this.streamTableInfo.addField(streamTableRowInfo.getFieldNames()[i]);
				this.streamTableInfo.addFieldTypeInfomation(streamTableRowInfo.getTypeAt(i));
			}
			this.streamTableInfo.finish();
		}
	}

	public XjoinMixStreamV4 registStreamTableTypeInfo(RowTypeInfo streamTableRowInfo) {

		for (int i = 0; i < streamTableRowInfo.getArity(); i++) {
			this.streamTableInfo.addField(streamTableRowInfo.getFieldNames()[i]);
			this.streamTableInfo.addFieldTypeInfomation(streamTableRowInfo.getTypeAt(i));
		}
		return this;
	}

	public ProjectStreamTable projectStreamTableKey(int... streamTableKeyPos) {
		if (this.streamTableInfo == null) {
			throw new IllegalArgumentException("streamTable must be row type");
		}
		return new ProjectStreamTable(streamTableKeyPos);
	}

	public class ProjectStreamTable {
		int[] streamTableKeyPos;
		int[] streamTableFieldPos;
		RowTypeInfo streamTableKeyTypeInfo;

		ProjectStreamTable(int... streamTableKeyPos) {
			this.streamTableKeyPos = streamTableKeyPos;
			this.streamTableFieldPos = projectPosition(streamTableInfo.getFieldList().size());
			this.streamTableKeyTypeInfo = projectFields(
				streamTableInfo.getFieldTypeInfomationList(),
				streamTableInfo.getFieldList(),
				this.streamTableKeyPos);

		}

		public ProjectStreamTable projectStreamTableField(int... streamTableFieldPos) {
			if (!isSubSets(this.streamTableFieldPos, streamTableFieldPos)) {
				throw new IllegalArgumentException("project fields out of range");
			}
			if (!isSubSets(streamTableFieldPos, streamTableKeyPos)) {
				throw new IllegalArgumentException("project fields must have key");
			}
			this.streamTableFieldPos = streamTableFieldPos;
			return this;
		}

		public ProjectSideTable ProjectSideTableKey(int... sideTableKeyPos) {
			return new ProjectSideTable(sideTableKeyPos);
		}

		public class ProjectSideTable {
			int[] sideTableKeyPos;
			int[] sideTableFieldPos;
			RowTypeInfo sideTableKeyTypeInfo;

			ProjectSideTable(int... sideTableKeyPos) {
				this.sideTableKeyPos = sideTableKeyPos;
				this.sideTableFieldPos = projectPosition(sideTableInfo.getFieldList().size());
				this.sideTableKeyTypeInfo = projectFields(
					sideTableInfo.getFieldTypeInfomationList(),
					sideTableInfo.getFieldList(),
					this.sideTableKeyPos);

			}

			public ProjectSideTable projectsideTableField(int... sideTableFieldPos) {
				if (!isSubSets(this.sideTableFieldPos, sideTableFieldPos)) {
					throw new IllegalArgumentException("project fields out of range");
				}
				if (!isSubSets(sideTableFieldPos, sideTableKeyPos)) {
					throw new IllegalArgumentException("project fields must have key");
				}
				this.sideTableFieldPos = sideTableFieldPos;
				return this;
			}

			public AsyncJoinInfo AsyncXjoin() {
				return new AsyncJoinInfo(XjoinMethod.ASYNC);
			}

			public SyncJoinInfo SyncXjoin() {
				return new SyncJoinInfo(XjoinMethod.SYNC);
			}

			public class AsyncJoinInfo {
				private final XjoinMethod xjoinMethod;
				private XjoinType xjoinType;
				private AsyncXjoinCacheType asyncXjoinCacheType;

				private int cacheSize = 10000;
				private long cacheTimeout = 60 * 1000;//
				private int asyncCapacity = 100;
				private int parallelism = 1;

				public AsyncJoinInfo(XjoinMethod xjoinMethod) {
					this.xjoinMethod = xjoinMethod;
					this.xjoinType = XjoinType.LEFT;
					this.asyncXjoinCacheType = AsyncXjoinCacheType.NONE;
				}

				public AsyncJoinInfo setXjoinType(XjoinType xjoinType) {
					this.xjoinType = xjoinType;
					return this;
				}

				public AsyncJoinInfo setAsyncXjoinCacheType(AsyncXjoinCacheType asyncXjoinCacheType) {
					this.asyncXjoinCacheType = asyncXjoinCacheType;
					return this;
				}

				public AsyncJoinInfo setCacheSize(int cacheSize) {
					this.cacheSize = cacheSize;
					return this;
				}

				public AsyncJoinInfo setCacheTimeout(long cacheTimeout) {
					this.cacheTimeout = cacheTimeout;
					return this;
				}

				public AsyncJoinInfo setAsyncCapacity(int asyncCapacity) {
					this.asyncCapacity = asyncCapacity;
					return this;
				}

				public AsyncJoinInfo setParallelism(int parallelism) {
					this.parallelism = parallelism;
					return this;
				}

				public SingleOutputMixStreamOperator<Row> apply() throws Exception {

					// 构建缓存\连接信息
					JoinInfo joinInfo = new JoinInfo();
					joinInfo.setAsyncCapacity(this.asyncCapacity);
					joinInfo.setCacheSize(this.cacheSize);
					joinInfo.setCacheTimeout(this.cacheTimeout);
					joinInfo.setAsyncCacheType(this.asyncXjoinCacheType);
					joinInfo.setParallelism(parallelism);

					joinInfo.setStreamTableName(streamTableInfo.getName());
					joinInfo.setSideTableName(sideTableInfo.getName());
					joinInfo.setXjoinType(xjoinType);


					//make outField
					outFieldInfoList = new ArrayList<>();
					for (int i : streamTableFieldPos) {
						FieldInfo fieldInfo = new FieldInfo();
						if (isSubSets(streamTableKeyPos, i)) {
							fieldInfo.setJoinKey(true);
						}
						fieldInfo.setTable(streamTableInfo.getName());
						fieldInfo.setFieldName(streamTableInfo.getFieldList().get(i));
						fieldInfo.setFieldPos(i);
						fieldInfo.setTypeInformation(streamTableInfo.getFieldTypeInfomationList().get(i));
						outFieldInfoList.add(fieldInfo);
					}
					for (int j : sideTableFieldPos) {
						FieldInfo fieldInfo = new FieldInfo();
						if (isSubSets(sideTableKeyPos, j)) {
							fieldInfo.setJoinKey(true);
						}
						fieldInfo.setTable(sideTableInfo.getName());
						fieldInfo.setFieldName(sideTableInfo.getFieldList().get(j));
						fieldInfo.setFieldPos(j);
						fieldInfo.setTypeInformation(sideTableInfo.getFieldTypeInfomationList().get(j));
						outFieldInfoList.add(fieldInfo);
					}

					//make return type
					ArrayList<TypeInformation<?>> typeInformations = new ArrayList<>();
					ArrayList<String> strings = new ArrayList<>();
					for (FieldInfo fieldInfo : outFieldInfoList) {
						if (fieldInfo.getTable().equalsIgnoreCase(streamTableInfo.getName())) {
							typeInformations.add(fieldInfo.getTypeInformation());
							strings.add(fieldInfo.getFieldName());
						} else if (fieldInfo.getTable().equalsIgnoreCase(sideTableInfo.getName())) {
							if (!fieldInfo.isJoinKey()) {
								typeInformations.add(fieldInfo.getTypeInformation());
								strings.add(fieldInfo.getFieldName());
							}
						} else {
							throw new IllegalArgumentException("wrong table");
						}
					}

					RowTypeInfo returnTypeInfo = new RowTypeInfo(
						typeInformations.toArray(new TypeInformation[typeInformations.size()]),
						strings.toArray(new String[strings.size()])
					);
					System.out.println(returnTypeInfo);
					//create operate
					return AsyncXjoinOperatorGateWay.getAsyncXjoinStream(
						streamTable,
						streamTableInfo,
						sideTableInfo,
						joinInfo,
						outFieldInfoList,
						returnTypeInfo)
						.setParallelism(joinInfo.getParallelism())
						.returns(returnTypeInfo);

				}
			}

			public class SyncJoinInfo {
				private XjoinMethod xjoinMethod;
				private XjoinType xjoinType;

				private int cacheSize = 10000;
				private long cacheTimeout = 10 * 60 * 1000;//
				private int parallelism = 1;

				private boolean isPartition = false;
				private boolean incLog = false;

				private String shareSlot="";

				SyncJoinInfo(XjoinMethod xjoinMethod) {
					this.xjoinMethod = xjoinMethod;
					this.xjoinType = XjoinType.LEFT;
				}

				public SyncJoinInfo setShareSlot(String shareSlot) {
					this.shareSlot = shareSlot;
					return this;
				}

				public SyncJoinInfo setXjoinType(XjoinType xjoinType) {
					this.xjoinType = xjoinType;
					return this;
				}

				public SyncJoinInfo enablePartition() {
					return enablePartition(true);
				}

				public SyncJoinInfo enablePartition(boolean isPartition) {
					this.isPartition = isPartition;
					return this;
				}

				public SyncJoinInfo enableIncLog() {
					return enableIncLog(true);
				}

				public SyncJoinInfo enableIncLog(boolean incLog) {
					this.incLog = incLog;
					return this;
				}


				public SyncJoinInfo setCacheSize(int cacheSize) {
					this.cacheSize = cacheSize;
					return this;
				}

				public SyncJoinInfo setCacheTimeout(long cacheTimeout) {
					this.cacheTimeout = cacheTimeout;
					return this;
				}

				public SyncJoinInfo setParallelism(int parallelism) {
					this.parallelism = parallelism;
					return this;
				}

				public SingleOutputMixStreamOperator<Row> apply() throws Exception {

					// 构建缓存\连接信息
					JoinInfo joinInfo = new JoinInfo();
					joinInfo.setCacheSize(this.cacheSize);
					joinInfo.setCacheTimeout(this.cacheTimeout);
					joinInfo.setPartitionedJoin(this.isPartition);
					joinInfo.setParallelism(parallelism);

					joinInfo.setStreamTableName(streamTableInfo.getName());
					joinInfo.setSideTableName(sideTableInfo.getName());
					joinInfo.setXjoinType(xjoinType);

					joinInfo.setIncLog(incLog);

					//make outField
					outFieldInfoList = new ArrayList<>();
					for (int i : streamTableFieldPos) {
						FieldInfo fieldInfo = new FieldInfo();
						if (isSubSets(streamTableKeyPos, i)) {
							fieldInfo.setJoinKey(true);
						}
						fieldInfo.setTable(streamTableInfo.getName());
						fieldInfo.setFieldName(streamTableInfo.getFieldList().get(i));
						fieldInfo.setFieldPos(i);
						fieldInfo.setTypeInformation(streamTableInfo.getFieldTypeInfomationList().get(i));
						outFieldInfoList.add(fieldInfo);
					}
					for (int j : sideTableFieldPos) {
						FieldInfo fieldInfo = new FieldInfo();
						if (isSubSets(sideTableKeyPos, j)) {
							fieldInfo.setJoinKey(true);
						}
						fieldInfo.setTable(sideTableInfo.getName());
						fieldInfo.setFieldName(sideTableInfo.getFieldList().get(j));
						fieldInfo.setFieldPos(j);
						fieldInfo.setTypeInformation(sideTableInfo.getFieldTypeInfomationList().get(j));
						outFieldInfoList.add(fieldInfo);
					}

					//make return type
					ArrayList<TypeInformation<?>> typeInformations = new ArrayList<>();
					ArrayList<String> strings = new ArrayList<>();
					for (FieldInfo fieldInfo : outFieldInfoList) {
						if (fieldInfo.getTable().equalsIgnoreCase(streamTableInfo.getName())) {
							typeInformations.add(fieldInfo.getTypeInformation());
							strings.add(fieldInfo.getFieldName());
						} else if (fieldInfo.getTable().equalsIgnoreCase(sideTableInfo.getName())) {
							if (!fieldInfo.isJoinKey()) {
								typeInformations.add(fieldInfo.getTypeInformation());
								strings.add(fieldInfo.getFieldName());
							}
						} else {
							throw new IllegalArgumentException("wrong table");
						}
					}

					RowTypeInfo returnTypeInfo = new RowTypeInfo(
						typeInformations.toArray(new TypeInformation[typeInformations.size()]),
						strings.toArray(new String[strings.size()])
					);
					System.out.println(returnTypeInfo);
					//create operate

					return SyncXjoinOperatorGateWay.getSyncXjoinStream(
						environment, streamTable,
						streamTableInfo, sideTableInfo, joinInfo,
						outFieldInfoList, returnTypeInfo,
						shareSlot,streamTableKeyPos)
						.setParallelism(joinInfo.getParallelism())
						.returns(returnTypeInfo);

				}
			}
		}
	}

	public static RowTypeInfo projectFields(List<TypeInformation> typeInfo, List<String> fieldName, int[] fieldMapping) {
		TypeInformation[] fieldTypes = new TypeInformation[fieldMapping.length];
		String[] fieldNames = new String[fieldMapping.length];
		for (int i = 0; i < fieldMapping.length; i++) {
			fieldTypes[i] = typeInfo.get(fieldMapping[i]);
			fieldNames[i] = fieldName.get(fieldMapping[i]);
		}
		return new RowTypeInfo(fieldTypes, fieldNames);
	}

	private static int[] projectPosition(int l) {
		int[] temp = new int[l];
		for (int i = 0; i < l; i++) {
			temp[i] = i;
		}
		return temp;
	}

	private static boolean isSubSets(int[] father, int... child) {
		boolean isSubSet = true; //null as subset
		for (int i : child) {
			isSubSet = false;
			for (int j : father) {
				if (j == i) {
					isSubSet = true;
					break;
				}
			}
			if (!isSubSet) {
				break;
			}
		}
		return isSubSet;
	}


}